package xubo.wangcaifeng.love.log

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{DataFrame, Row, SQLContext}


object DataFrameDataDeal {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("log")
      .setMaster("local")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
      .set("spark.sql.parquet.compression.codec", "Snappy")
    val sc = new SparkContext(conf)
    val sqlc = new SQLContext(sc)
    val lines: RDD[String] = sc.textFile("data/2016-10-01_06_p1_invalid.1475274123982.log.FINISH.bz2")
    //lines.foreach(println)
    val filterData: RDD[Array[String]] = lines.map(_.split(",")).filter(_.length == 85)
    //filterData.foreach(i=>println(i.length))
    val logData: RDD[Row] = filterData.map(t => {
      val invalid = "-1"
      val sessionid = t(0)
      val advertisersid = (if (t(1)=="") invalid else t(1)).toInt
      val adorderid = (if (t(2)=="") invalid else t(2)).toInt
      val adcreativeid = (if (t(3)=="") invalid else t(3)).toInt
      val adplatformproviderid = (if (t(4)=="") invalid else t(4)).toInt
      val sdkversion = t(5)
      val adplatformkey = t(6)
      val putinmodeltype = (if (t(7)=="") invalid else t(7)).toInt
      val requestmode = (if (t(8)=="") invalid else t(8)).toInt
      val adprice = t(9)
      val adppprice = t(10)
      val requestdate = t(11)
      val ip = t(12)
      val appid = t(13)
      val appname = t(14)
      val uuid = t(15)
      val device = t(16)
      val client = (if (t(17)=="") invalid else t(17)).toInt
      val osversion = t(18)
      val density = t(19)
      val pw = (if (t(21)=="") invalid else t(20)).toInt
      val ph = (if (t(21)=="") invalid else t(21)).toInt
      val long = t(22)
      val lat = t(23)
      val provincename = t(24)
      val cityname = t(25)
      val ispid = (if (t(26)=="") invalid else t(26)).toInt
      val ispname = t(27)
      val networkmannerid = (if (t(28)=="") invalid else t(28)).toInt
      val networkmannername = t(29)
      val iseffective = (if (t(30)=="") invalid else t(30)).toInt
      val isbilling = (if (t(31)=="") invalid else t(31)).toInt
      val adspacetype = (if (t(32)=="") invalid else t(32)).toInt
      val adspacetypename = t(33)
      val devicetype = (if (t(34)=="") invalid else t(34)).toInt
      val processnode = (if (t(35)=="") invalid else t(35)).toInt
      val apptype = (if (t(36)=="") invalid else t(36)).toInt
      val district = t(37)
      val paymode = (if (t(38)=="") invalid else t(38)).toInt
      val isbid = (if (t(39)=="") invalid else t(39)).toInt
      val bidprice = (if(t(40)=="") invalid else t(40)).toDouble
      val winprice = (if (t(41)=="") invalid else t(41)).toDouble
      val iswin = (if (t(42)=="") invalid else t(42)).toInt
      val cur = t(43)
      val rate = (if (t(44)=="") invalid else t(44)).toDouble
      val cnywinprice = (if(t(45)=="") invalid else t(45)).toDouble
      val imei = t(46)
      val mac = t(47)
      val idfa = t(48)
      val openudid = t(49)
      val androidid = t(50)
      val rtbprovince = t(51)
      val rtbcity = t(52)
      val rtbdistrict = t(53)
      val rtbstreet = t(54)
      val storeurl = t(55)
      val realip = t(56)
      val isqualityapp = (if (t(57)=="") invalid else t(57)).toInt
      val bidfloor = (if (t(58)=="") invalid else t(58)).toDouble
      val aw = (if (t(59)=="") invalid else t(59)).toInt
      val ah = (if (t(60)=="") invalid else t(60)).toInt
      val imeimd5 = t(61)
      val macmd5 = t(62)
      val idfamd5 = t(63)
      val openudidmd5 = t(64)
      val androididmd5 = t(65)
      val imeisha1 = t(66)
      val macsha1 = t(67)
      val idfasha1 = t(68)
      val openudidsha1 = t(69)
      val androididsha1 = t(70)
      val uuidunknow = t(71)
      val userid = t(72)
      val iptype = (if (t(73)=="") invalid else t(73)).toInt
      val initbidprice = (if (t(74)=="") invalid else t(74)).toDouble
      val adpayment = (if (t(75)=="") invalid else t(75)).toDouble
      val agentrate = (if (t(76)=="") invalid else t(76)).toDouble
      val lomarkrate = (if (t(77)=="") invalid else t(77)).toDouble
      val adxrate = (if (t(78)=="") invalid else t(78)).toDouble
      val title = t(79)
      val keywords = t(80)
      val tagid = t(81)
      val callbackdate = t(82)
      val channelid = t(83)
      val mediatype = (if (t(84)=="") invalid else t(84)).toInt
      //将所有的信心封装成Row
      Row(
        sessionid,
        advertisersid,
        adorderid,
        adcreativeid,
        adplatformproviderid,
        sdkversion,
        adplatformkey,
        putinmodeltype,
        requestmode,
        adprice,
        adppprice,
        requestdate,
        ip,
        appid,
        appname,
        uuid,
        device,
        client,
        osversion,
        density,
        pw,
        ph,
        long,
        lat,
        provincename,
        cityname,
        ispid,
        ispname,
        networkmannerid,
        networkmannername,
        iseffective,
        isbilling,
        adspacetype,
        adspacetypename,
        devicetype,
        processnode,
        apptype,
        district,
        paymode,
        isbid,
        bidprice,
        winprice,
        iswin,
        cur,
        rate,
        cnywinprice,
        imei,
        mac,
        idfa,
        openudid,
        androidid,
        rtbprovince,
        rtbcity,
        rtbdistrict,
        rtbstreet,
        storeurl,
        realip,
        isqualityapp,
        bidfloor,
        aw,
        ah,
        imeimd5,
        macmd5,
        idfamd5,
        openudidmd5,
        androididmd5,
        imeisha1,
        macsha1,
        idfasha1,
        openudidsha1,
        androididsha1,
        uuidunknow,
        userid,
        iptype,
        initbidprice,
        adpayment,
        agentrate,
        lomarkrate,
        adxrate,
        title,
        keywords,
        tagid,
        callbackdate,
        channelid,
        mediatype)
    })
    //自定义schema信息
    val logSchema: StructType = StructType(
      List(
        StructField("sessionid",StringType),
        StructField("advertisersid",IntegerType),
        StructField("adorderid",IntegerType),
        StructField("adcreativeid",IntegerType),
        StructField("adplatformproviderid",IntegerType),
        StructField("sdkversion",StringType),
        StructField("adplatformkey",StringType),
        StructField("putinmodeltype",IntegerType),
        StructField("requestmode",IntegerType),
        StructField("adprice",DoubleType),
        StructField("adppprice",DoubleType),
        StructField("requestdate",StringType),
        StructField("ip",StringType),
        StructField("appid",StringType),
        StructField("appname",StringType),
        StructField("uuid",StringType),
        StructField("device",StringType),
        StructField("client",IntegerType),
        StructField("osversion",StringType),
        StructField("density",StringType),
        StructField("pw",IntegerType),
        StructField("ph",IntegerType),
        StructField("long",StringType),
        StructField("lat",StringType),
        StructField("provincename",StringType),
        StructField("cityname",StringType),
        StructField("ispid",IntegerType),
        StructField("ispname",StringType),
        StructField("networkmannerid",IntegerType),
        StructField("networkmannername",StringType),
        StructField("iseffective",IntegerType),
        StructField("isbilling",IntegerType),
        StructField("adspacetype",IntegerType),
        StructField("adspacetypename",StringType),
        StructField("devicetype",IntegerType),
        StructField("processnode",IntegerType),
        StructField("apptype",IntegerType),
        StructField("district",StringType),
        StructField("paymode",IntegerType),
        StructField("isbid",IntegerType),
        StructField("bidprice",DoubleType),
        StructField("winprice",DoubleType),
        StructField("iswin",IntegerType),
        StructField("cur",StringType),
        StructField("rate",DoubleType),
        StructField("cnywinprice",DoubleType),
        StructField("imei",StringType),
        StructField("mac",StringType),
        StructField("idfa",StringType),
        StructField("openudid",StringType),
        StructField("androidid",StringType),
        StructField("rtbprovince",StringType),
        StructField("rtbcity",StringType),
        StructField("rtbdistrict",StringType),
        StructField("rtbstreet",StringType),
        StructField("storeurl",StringType),
        StructField("realip",StringType),
        StructField("isqualityapp",IntegerType),
        StructField("bidfloor",DoubleType),
        StructField("aw",IntegerType),
        StructField("ah",IntegerType),
        StructField("imeimd5",StringType),
        StructField("macmd5",StringType),
        StructField("idfamd5",StringType),
        StructField("openudidmd5",StringType),
        StructField("androididmd5",StringType),
        StructField("imeisha1",StringType),
        StructField("macsha1",StringType),
        StructField("idfasha1",StringType),
        StructField("openudidsha1",StringType),
        StructField("androididsha1",StringType),
        StructField("uuidunknow",StringType),
        StructField("userid",StringType),
        StructField("iptype",IntegerType),
        StructField("initbidprice",DoubleType),
        StructField("adpayment",DoubleType),
        StructField("agentrate",DoubleType),
        StructField("lomarkrate",DoubleType),
        StructField("adxrate",DoubleType),
        StructField("title",StringType),
        StructField("keywords",StringType),
        StructField("tagid",StringType),
        StructField("callbackdate",StringType),
        StructField("channelid",StringType),
        StructField("mediatype",IntegerType)
      )
    )
    val logDF: DataFrame = sqlc.createDataFrame(logData,logSchema)
    logDF.coalesce(1).write.parquet("data/parquet")
    sc.stop()

  }

}
